package com.zhike.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.zhike.bean.AlarmTypeBean;
import com.zhike.function.DimAsyncFunction;
import com.zhike.utils.DateFormatUtil;
import com.zhike.utils.DorisUtils;
import com.zhike.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhike
 * @CreateTime: 2024/2/26 13:47
 * @Description: 计算各车型,品牌的告警信息并写入Doris
 */
public class DwsAlarmTypeBrandWindow {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(2);

        //2.读取故障告警日志创建流并转换数据结构
        SingleOutputStreamOperator<JSONObject> alarmDS = env.fromSource(KafkaUtils.getKafkaSource("car_alarm_log", "dws_alarm_window"),
                        WatermarkStrategy.noWatermarks(), "alarm_log_source")
                .map(JSONObject::parseObject);

        //3.将数据流封装为AlarmTypeBean对象
        SingleOutputStreamOperator<AlarmTypeBean> alarmBeanDS = alarmDS.map(new MapFunction<JSONObject, AlarmTypeBean>() {
            @Override
            public AlarmTypeBean map(JSONObject value) throws Exception {
                //获取vin
                String vin = value.getString("vin");
                //获取告警级别
                Integer alarmLevel = value.getInteger("alarm_level");
                //获取时间戳
                Long ts = value.getLong("timestamp");
                return AlarmTypeBean.builder()
                        .vin(vin)
                        .alarmCount(1)
                        .alarmCountL1(alarmLevel == 1 ? 1 : 0)
                        .alarmCountL2(alarmLevel == 2 ? 1 : 0)
                        .alarmCountL3(alarmLevel == 3 ? 1 : 0)
                        .ts(ts)
                        .build();
            }
        });

        //4.关联car_info维表，补充车型、品牌字段
        SingleOutputStreamOperator<AlarmTypeBean> rishDS = AsyncDataStream.unorderedWait(alarmBeanDS, new DimAsyncFunction<AlarmTypeBean>("DIM_CAR_INFO") {
                    @Override
                    public String getKey(AlarmTypeBean input) {
                        return input.getVin();
                    }

                    @Override
                    public void join(AlarmTypeBean input, JSONObject dimInfo) {
                        input.setCarType(dimInfo.getString("TYPE"));
                        input.setBrand(dimInfo.getString("TRADEMARK"));
                    }
                }
                , 1000, TimeUnit.SECONDS);

        //5.设置水位线
        SingleOutputStreamOperator<AlarmTypeBean> waterMarkDS = rishDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<AlarmTypeBean>forBoundedOutOfOrderness(Duration.ofSeconds(30L))
                .withTimestampAssigner(new SerializableTimestampAssigner<AlarmTypeBean>() {
                    @Override
                    public long extractTimestamp(AlarmTypeBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        //6.按照车型品牌分组、开窗、聚合
        SingleOutputStreamOperator<AlarmTypeBean> reduceDS = waterMarkDS.keyBy(new KeySelector<AlarmTypeBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(AlarmTypeBean value) throws Exception {
                        return new Tuple2<>(value.getCarType(), value.getBrand());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(60L)))
                .reduce(new ReduceFunction<AlarmTypeBean>() {
                    @Override
                    public AlarmTypeBean reduce(AlarmTypeBean value1, AlarmTypeBean value2) throws Exception {
                        value1.setAlarmCount(value1.getAlarmCount() + value2.getAlarmCount());
                        value1.setAlarmCountL1(value1.getAlarmCountL1() + value2.getAlarmCountL1());
                        value1.setAlarmCountL2(value1.getAlarmCountL2() + value2.getAlarmCountL2());
                        value1.setAlarmCountL3(value1.getAlarmCountL3() + value2.getAlarmCountL3());
                        return value1;
                    }
                }, new WindowFunction<AlarmTypeBean, AlarmTypeBean, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow window, Iterable<AlarmTypeBean> input, Collector<AlarmTypeBean> out) throws Exception {
                        AlarmTypeBean bean = input.iterator().next();
                        bean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        bean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        bean.setWindowDate(DateFormatUtil.toDate(window.getStart()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });

        reduceDS.print();
        //7.将结果转换为json字符串并写入doris
        SingleOutputStreamOperator<String> resultDS = reduceDS.map(new MapFunction<AlarmTypeBean, String>() {
            @Override
            public String map(AlarmTypeBean value) throws Exception {
                SerializeConfig serializeConfig = new SerializeConfig();
                serializeConfig.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                return JSON.toJSONString(value, serializeConfig, SerializerFeature.WriteMapNullValue);
            }
        });

//        resultDS.print();
        resultDS.sinkTo(DorisUtils.getDorisSink("dws_alarm_bt_window"));

        //8.启动任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
